/**!
并行计算相关的简陋尝试。
*/
use serde::{Deserialize, Serialize};
#[cfg(feature = "tracelog")]
use tracing::error;
#[cfg(feature = "rotatelog")]
use log::{debug, error, info, warn};

use anyhow::bail;

use reqwest::header::HeaderMap;
use std::collections::HashMap;
use std::borrow::Cow;
use std::path::PathBuf;
use crate::*;


/**
REST请求中， 传递算子信息的结构

*/ 
#[derive(Serialize, Deserialize, Debug, Clone, Default)]
pub struct ParallelArgs {
    ///算子名称
    #[serde(default = "default_string")]
    pub libname: String,
    #[serde(default = "default_string")]
    pub token: String,
    ///算子在gfs上的idx
    #[serde(default = "default_number")]        
    pub idx: i32,
    /// 用于执行算子时的起始ihash
    #[serde(default = "default_number")]    
    pub min: i32,
    /// 用于执行算子时的终止ihash
    #[serde(default = "default_number")]    
    pub max: i32,
}

/**
算子的全域（根据router）部署
 - 算子将被写到所有节点的Master（自动同步到slave），并返回各个节点的定位信息（HashMap<(i32,i32), GfsResponse>）
 - 返回的定位信息，各个Node的ihash范围值，元信息（min,max)作为key。
 - 此返回值将作为发起计算的算子定位信息。
 
 */
pub async fn gfsc_operator_deploy(router: &GfsRouter,  file_name: &str) -> Result<  HashMap<(i32,i32), GfsResponse> > {    

    let filename = PathBuf::from(  file_name);
    if !filename.exists() {
        bail!("no such file");
    }
    // get all node's writer by router    
    let writers = router.get_dynasty_writers( router.cur_dynasty)?;
    let only_fname = filename
        .file_name()
        .unwrap_or_default()
        .to_str()
        .unwrap_or_default();
    
    let file_byte = std::fs::read(PathBuf::from(file_name))?;
    // 一定要有file_name方法，且参数不能为空，否则数据上传失败
    let file_name = only_fname.to_string(); //filename.file_name().unwrap_or_default().to_string_lossy().to_string();
    let mut retv : HashMap<(i32, i32), GfsResponse> = HashMap::new();
    for ((min, max), writer) in writers.iter() {
    // for i in 0..writers.len() {                
        let token = new_token();               
        let gfs_writer_url = format!("{}/write/file/{}", writer , token);
        let part = reqwest::multipart::Part::bytes(Cow::from(file_byte.clone())).file_name(file_name.clone());
        let r = write_to_gfs_with_part(gfs_writer_url.as_str(), only_fname, part).await?;
        //let key = format!("{}", node);
        //retv.insert(key, r);  
        let ikey = (*min, *max);             
        retv.insert(ikey, r);                
    }        
    // write file to each writer
    return Ok(retv);    
}

/**
检测算子是否存在（输入信息是部署算子时的返回值）
 - 输入信息HashMap<(i32, i32), GfsResponse > 不能随便创作，其key必须与router中的节点的(min,max)信息对应 
 */

pub async fn gfsc_operator_exists(router: &GfsRouter, operator_tables: &HashMap<(i32, i32), GfsResponse > ) -> Result<()> {    
    // get all node's reader by router
    let readers = router.get_dynasty_readers(router.cur_dynasty)?;
    for (node, res) in operator_tables{
        let reader = readers.get(node);
        if let Some(reader) = reader {
            gfsc_file_exist(reader, res.ihash, res.idx, &res.name).await?;
        }else{
            bail!( format!("{:#?} is not exist in router", node))
        }
    }    
    return Ok(());
}

/**
根据提供的算子idx信息ids: HashMap<(i32,i32),i32>，从每个节点中获取一台服务器，执行指定算子的计算服务。
 - 算子ids的key，必须与router中的节点min,max保持一致。也是与部署算子时返回的值保持一致。
 - 参与计算的服务器，在本系统中是随机从队列中获取的。 
 */
pub async fn gfsc_map_task( router: &GfsRouter, libname: &str,ids: HashMap<(i32,i32),i32>, args: &mut HashMap<String, String> ) -> Result< () > {
    let readers = router.get_dynasty_readers(router.cur_dynasty)?;
    // let ihash = string_hash256(libname);
    for ((min, max), reader) in readers.iter() {
        let id_key = (*min, *max);
        let idx = ids.get(&id_key);
        if let Some(idx) = idx {                        
            let token = new_token();               
            let task_url = format!("{}/parallel/task?token={}&libname={}&idx={}&min={}&max={}",
                reader, token, urlencoding::encode(libname), *idx, *min, *max);
            invoke_gfs_task(&task_url, args).await?;     
        }else{
            error!( "node-{:#?} is not in router, this node no task", &id_key);
        }
    }
    return Ok(());
}

/**
执行特定服务器上的特定算子的计算服务。
 - 机器是指定的（为何是他，需要外部清楚）
 - task_url构造了完整的调用信息，包括算子的信息。
 - 此函数是gfsc_map_task的基础，同时也可以独立构造出更广泛的计算模型。
   * gfsc_map_task中，一个节点只能有一台服务器参与计算。
   * 基于此函数，根据Node中的服务器数量，通过分解（min,max), 可以发起更大范围的任务分发，从而实现所有服务资源的利用。
   * 使用此函数的前提是，充分理解计算模型和规则。
 */
pub async fn invoke_gfs_task( task_url: &str, args: &HashMap<String, String> ) -> Result< () > {
    let client = reqwest::Client::new();
    let mut headers = HeaderMap::new();
    headers.insert("Content-Type", "application/json; charset=utf-8".parse()?);
    headers.insert("Authorization", format!("Bearer {}", "token").parse()?);
    let r = client
        .post(task_url)
        .headers(headers)
        .json(&args)
        .send()
        .await?
        .text().await?;   
    if r.eq("OK") || r.eq("ok"){
        return Ok(())
    }
    bail!("invoke error")         
}

/**
获得特定算子名称在系统中的record文件名称。
 - 为了防止名称冲突，特意统一进行了分布式计算的记录文件命名 
 */
pub fn gfsc_parallel_record_name(libname: &str)-> String {
    if filename_is_valid(libname){
        return format!("{}.record", libname)
    }else{
        return format!("gfs_parallel.record")
    }
}